package com.nl.market;

import com.nl.bean.input.AdClickEvent;
import com.nl.utils.WindowAggResultOut;
import com.nl.utils.aggfunctions.CountAgg;
import java.sql.Timestamp;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author shihb
 * @date 2019/12/20 18:28
 * 按省份统计广告点击量，一个用户点击同一个广告的上限为100，
 * 超过加入黑名单，之后点击该广告无效。明天凌晨重置黑名单
 */
public class AdClickStatistical {

  // 定义黑名单的测输出流
  private static OutputTag<String> blacklistTag = new OutputTag<String>("blacklist") {
  };

  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    env.setParallelism(1);
    DataStreamSource<String> source = env.readTextFile(
        AdClickStatistical.class.getClassLoader().getResource("AdClickLog.csv").getPath());
    SingleOutputStreamOperator<AdClickEvent> adEventStream = source
        .map(s -> {
          String[] arr = s.split(",");
          long userId = Long.parseLong(arr[0].trim());
          long adId = Long.parseLong(arr[1].trim());
          String province = arr[2].trim();
          String city = arr[3].trim();
          long timestamp = Long.parseLong(arr[4].trim());
          return new AdClickEvent(userId, adId, province, city, timestamp);
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<AdClickEvent>(
            Time.seconds(0)) {
          @Override
          public long extractTimestamp(AdClickEvent element) {
            return element.getTimestamp() * 1000;
          }
        });

    //定义process function 过滤点击量大得行为
    SingleOutputStreamOperator<AdClickEvent> filterBlacklistStream = adEventStream
        .keyBy("userId", "adId")
        .process(new BlacklistFunction(100));

    //根据省份做分组,开窗聚合
    SingleOutputStreamOperator<String> adCountStream = filterBlacklistStream
        // 按照
        .keyBy(adClickEvent->adClickEvent.getProvince())
        .timeWindow(Time.hours(1), Time.minutes(5))
        .aggregate(new CountAgg<AdClickEvent>(), new WindowAggResultOut<String>())
        .map(input -> {
          return new Timestamp(input.getWindowEnd()).toString() + " " + input.getKey() + "点击量:" + input.getCount();
        });


    //黑名单侧输出流
    DataStream<String> blacklistStream = filterBlacklistStream.getSideOutput(blacklistTag);

    adCountStream.print("count");
    blacklistStream.print("blacklist");
    env.execute("ad clinck job");


  }


  private static class BlacklistFunction extends
      KeyedProcessFunction<Tuple, AdClickEvent, AdClickEvent> {

    private int clickCap;
    private ValueState<Long> clickCount;
    private ValueState<Boolean> isSendBlacklist;
    private ValueState<Long> resetTimer;

    public BlacklistFunction(int clickCap) {
      this.clickCap=clickCap;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
      //定义状态，保存当前用户对当前广告得点击量
      clickCount=getRuntimeContext().getState(new ValueStateDescriptor<Long>("clickCount-state",Long.class));
      //保存黑名单发送的状态,让黑名单只输出一次
      isSendBlacklist=getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("sendBlacklist-state",Boolean.class));
      //保存定时器触发时间戳
      resetTimer=getRuntimeContext().getState(new ValueStateDescriptor<Long>("resetTimer-state",Long.class));

    }

    @Override
    public void processElement(AdClickEvent value, Context ctx, Collector<AdClickEvent> out)
        throws Exception {

      //获取点击次数
      Long curCount = clickCount.value();
      if(curCount==null){
        //第一次处理注册定时器，每天凌点重置状态
        //明天凌晨的时间戳
        Long ts=(ctx.timerService().currentProcessingTime()/(1000*60*60*24)+1)*(1000*60*60*24);
        // 注册定时器,用来清空状态
        ctx.timerService().registerProcessingTimeTimer(ts);
        resetTimer.update(ts);
        curCount=0L;
        isSendBlacklist.update(false);

      }else if(curCount>=clickCap){
        //大于点击上限如果没有发送黑名单，发送黑名单
        if(!isSendBlacklist.value()){
          String msg=value.getUserId()+" click "+value.getAdId()+" over "+clickCap + "times today";
          // 侧输出流
          ctx.output(blacklistTag,msg);
          // 输出完把输出状态设置为true
          isSendBlacklist.update(true);
        }
        // 大于上限就直接return主流不在输出
        return;
      }
      // 没有达到上限,更新点击数,主流正常输出
      curCount++;
      clickCount.update(curCount);
      out.collect(value);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<AdClickEvent> out)
        throws Exception {
      //定时器触发时清空状态
      if(timestamp==resetTimer.value()){
        this.isSendBlacklist.clear();
        this.clickCount.clear();
        this.resetTimer.clear();
      }
    }
  }

}

